package com.hgw.easyrocketmq.base.producer.impl;

import com.hgw.easyrocketmq.base.producer.BaseMsgProducer;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.Message;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.util.concurrent.TimeUnit;

/**
 * Description: 普通消息生产者
 * com.hgw.easyrocketmq.base.producer.impl->BaseMsgProducerImpl
 *
 * @author LinHuiBa-YanAn
 * @date 2023/1/2 13:09
 */
@Slf4j
@Service
public class BaseMsgProducerImpl implements BaseMsgProducer {

	@Value("${config.rocket-mq.name-server-addr}")
	private String nameSrvAddr;

	@Override
	public Boolean sendSyncMsg(String producerGroup, String topic, String tag, String context) {
		// 1. 创建消息生产者producer，并指定生产者组名
		DefaultMQProducer producer = new DefaultMQProducer(producerGroup);
		// 2. 指定NameServer地址
		producer.setNamesrvAddr(nameSrvAddr);
		// 3. 启动Producer
		try {
			producer.start();
		} catch (MQClientException e) {
			log.error("发送普通同步消息时，启动DefaultMQProducer失败，e:{}", e);
			e.printStackTrace();
		}
		// 4. 创建消息对象，指定主题Topic、Tag和消息体
		/**
		 * 参数一：消息主题Topic
		 * 参数二：消息Tag
		 * 参数三：消息内容
		 */
		Message msg = new Message(topic, tag, context.getBytes());
		// 5. 发送消息
		SendResult result = null;
		try {
			result = producer.send(msg);
		} catch (Exception e) {
			log.error("发送普通同步消息失败，e:{}", e);
			e.printStackTrace();
		}
		// 6. 关闭生产者 Producer
		producer.shutdown();
		// 发送状态
		SendStatus sendStatus = result.getSendStatus();
		if (sendStatus.equals(SendStatus.SEND_OK)) {
			log.info("发送成功,{}", result);
			return Boolean.TRUE;
		} else {
			return Boolean.FALSE;
		}
	}

	@Override
	public void sendAsyncMsg(String producerGroup, String topic, String tag, String context, SendCallback sendCallback) {
		// 1. 创建消息生产者producer，并指定生产者组名
		DefaultMQProducer producer = new DefaultMQProducer(producerGroup);
		// 2. 指定NameServer地址
		producer.setNamesrvAddr(nameSrvAddr);
		// 3. 启动Producer
		try {
			producer.start();
		} catch (MQClientException e) {
			log.error("发送异步消息时，启动DefaultMQProducer失败，e:{}", e);
			e.printStackTrace();
		}
		// 设置发送异步失败时的重试次数
		producer.setRetryTimesWhenSendAsyncFailed(3);
		// 4. 创建消息对象，指定主题Topic、Tag和消息体
		Message msg = new Message(topic, tag, context.getBytes());
		// 5. 发送消息
		try {
			producer.send(msg, sendCallback);
		} catch (Exception e) {
			log.error("发送异步消息失败，e:{}", e);
			e.printStackTrace();
		}
		producer.shutdown();
	}

	@Override
	public void sendOneWayMsg(String producerGroup, String topic, String tag, String context) {
		// 1. 创建消息生产者producer，并指定生产者组名
		DefaultMQProducer producer = new DefaultMQProducer(producerGroup);
		// 2. 指定NameServer地址
		producer.setNamesrvAddr(nameSrvAddr);
		// 3. 启动Producer
		try {
			producer.start();
		} catch (MQClientException e) {
			log.error("发送异步消息时，启动DefaultMQProducer失败，e:{}", e);
			e.printStackTrace();
		}
		// 设置发送异步失败时的重试次数
		producer.setRetryTimesWhenSendAsyncFailed(3);
		// 4. 创建消息对象，指定主题Topic、Tag和消息体
		Message msg = new Message(topic, tag, context.getBytes());
		// 5. 发送消息
		try {
			producer.sendOneway(msg);
			TimeUnit.SECONDS.sleep(1);
		} catch (Exception e) {
			log.error("发送异步消息失败，e:{}", e);
			e.printStackTrace();
		}
		producer.shutdown();
	}

}
